package tech.abing.spider.task.pipeline;

import java.io.File;
import java.io.FileWriter;
import java.io.IOException;
import java.util.List;

import org.apache.log4j.Logger;

import tech.abing.spider.component.Pipeline;
import tech.abing.spider.constants.FinalConstants;
import tech.abing.spider.model.Proccessable;
import tech.abing.spider.task.model.ComboInfo;
import tech.abing.spider.task.model.Product;
import tech.abing.spider.task.model.TaskTrade;
import tech.abing.spider.util.DateUtils;
import tech.abing.spider.util.ProccessableUtil;
import tech.abing.spider.util.RedisUtil;
import tech.abing.spider.util.SqlUtil;
import tech.abing.spider.util.StringUtils;
import tech.abing.spider.util.SysUtil;

public class ProductPipeline implements Pipeline {

	/**
	 * 
	 */
	private static final long serialVersionUID = 1L;
	
	private static Logger logger = Logger.getLogger(ProductPipeline.class);

	private static FileWriter writer = null;
	
	private static String fileName = "", path = "";
	
	private static SqlUtil sqlUtil = null;
	
	private static RedisUtil redisUtil = null;
	
	public ProductPipeline() {
		
		if(sqlUtil == null){
			//创建数据库链接
			sqlUtil = SqlUtil.getInstance();
			logger.info("数据库连接创建成功！");
		}
		
		if(redisUtil == null){
			redisUtil = RedisUtil.getInstance();
			logger.info("Redis连接池创建成功！");
		}
		
		fileName = "prd_"+DateUtils.getDate()+"_"+SysUtil.getUUID().replace("-", "")+".dat";
		logger.info("=========>输出产品的文件名："+fileName);
		path = FinalConstants.CRAWL_STORAGE_FOLDER_PREFIX+"TMALL/prd/"+DateUtils.getDate()+"/";
		
		File file = new File(path);
		if(!file.exists()){
			file.mkdirs();
		}
		file = new File(path+fileName);
		
		// 打开一个写文件器，构造函数中的第二个参数true表示以追加形式写文件
		try {
			if(writer == null){
				writer = new FileWriter(file, true);
			}
		} catch (IOException e) {
			e.printStackTrace();
		}
		
	}
	
	@Override
	public void proccessData(List<Proccessable> objectContainer) {
		
		if (ProccessableUtil.instanceOfClass(objectContainer, Product.class)) {// 判断集合类型是否是Product
			List<Product> list = ProccessableUtil.convert(objectContainer, Product.class);// 转换
			
			try {
				
				StringBuffer content = new StringBuffer();
				
				for(Product prod : list){
					
					if(StringUtils.isBlank(prod.getPrdName()) || (StringUtils.isBlank(prod.getPrice()) && (StringUtils.isBlank(prod.getProPrice())))){
						continue;//关键字段没采集到跳出循环
					}
					
					content.append(prod.getPrdNo()).append(FinalConstants.SPLIT).append(prod.getPrdName()).append(FinalConstants.SPLIT)
					.append(prod.getBrand()).append(FinalConstants.SPLIT).append(prod.getPrice()).append(FinalConstants.SPLIT)
					.append(prod.getProPrice()).append(FinalConstants.SPLIT).append(prod.getPromoType()).append(FinalConstants.SPLIT)
					.append(prod.getShopId()).append(FinalConstants.SPLIT).append(prod.getShopName()).append(FinalConstants.SPLIT)
					.append(prod.getSellerId()).append(FinalConstants.SPLIT).append(prod.getCat()).append(FinalConstants.SPLIT)
					.append(prod.getImgUri()).append(FinalConstants.SPLIT).append(prod.getMonthSale()).append(FinalConstants.SPLIT)
					.append(prod.getCollectNum()).append(FinalConstants.SPLIT).append(prod.getComments()).append(FinalConstants.SPLIT)
					.append(prod.getHasCombo()).append(FinalConstants.SPLIT).append(prod.getStart()).append(FinalConstants.SPLIT)
					.append(prod.getEnd()).append(FinalConstants.SPLIT).append(prod.getSellerAddr()).append(FinalConstants.SPLIT)
					.append(prod.getParam()).append(FinalConstants.SPLIT).append(prod.getCatchDate()).append(FinalConstants.SPLIT)
					.append(prod.getCoin()).append(FinalConstants.SPLIT).append(prod.getShopType()).append(FinalConstants.ENTER_CHAR);
					
					//增加交易采集任务
					int monthSale = StringUtils.toInteger(prod.getMonthSale());
					
					if(monthSale > 0){
						boolean exist = redisUtil.exist(prod.getPrdNo(), SysUtil.getValue("redis.unique_product_id"));
						
						if(!exist){//不存在则加入新的任务到队列中
							//TaskTrade prodForTrade = new TaskTrade(prod.getPrdNo(), prod.getShopId(), FinalConstants.SHOP_TYPE_TAMLL, DateUtils.getStartDateTime(), DateUtils.getStartDateTime());
							//TODO:第一天测试用,后续要改过来
							TaskTrade prodForTrade = new TaskTrade(prod.getPrdNo(), prod.getShopId(), prod.getShopType(), DateUtils.getLastStartDateTime(), DateUtils.getLastStartDateTime(), prod.getMonthSale());
							
							String unsaleTask = SysUtil.getValue("redis.task_trade_product_id_unsale");
							String normalTask = SysUtil.getValue("redis.task_trade_product_id_normal");
							String hotTask = SysUtil.getValue("redis.task_trade_product_id_hot");
							
							if(monthSale >= FinalConstants.PRODUCT_HOT_SALE_MIN){//热销
								
								//push到制定队列中
								try {
									redisUtil.lpush(prodForTrade, hotTask);
									logger.info("产品ID["+prodForTrade+"]入队队尾["+hotTask+"]成功");
								} catch (Exception e) {
									e.printStackTrace();
									logger.error("产品ID["+prodForTrade+"]入队队尾["+hotTask+"]失败", e);
								}
								
							}else if(monthSale >= FinalConstants.PRODUCT_NORMAL_SALE_MIN && monthSale < FinalConstants.PRODUCT_HOT_SALE_MIN){//普通
								
								//push到制定队列中
								try {
									redisUtil.lpush(prodForTrade, normalTask);
									logger.info("产品ID["+prodForTrade+"]入队队尾["+normalTask+"]成功");
								} catch (Exception e) {
									e.printStackTrace();
									logger.error("产品ID["+prodForTrade+"]入队队尾["+normalTask+"]失败", e);
								}
								
							}else if(monthSale > FinalConstants.PRODUCT_UNSALE_SALE_MIN && monthSale < FinalConstants.PRODUCT_NORMAL_SALE_MIN){//滞销
								
								//push到制定队列中
								try {
									redisUtil.lpush(prodForTrade, unsaleTask);
									logger.info("产品ID["+prodForTrade+"]入队队尾["+unsaleTask+"]成功");
								} catch (Exception e) {
									e.printStackTrace();
									logger.error("产品ID["+prodForTrade+"]入队队尾["+unsaleTask+"]失败", e);
								}
								
							}
							
						}
					}
				}
				
				writer.write(content.toString());
				writer.flush();
			} catch (IOException e) {
				logger.error("=========>离线写产品数据失败："+fileName);
			}
		}
		
		
		if(ProccessableUtil.instanceOfClass(objectContainer, ComboInfo.class)){
			List<ComboInfo> list = ProccessableUtil.convert(objectContainer, ComboInfo.class);// 转换
			
			sqlUtil.batchInsert(ComboInfo.class, list);
			
		}

	}

	@Override
	public void destory() {
		logger.info("=========>清理IO资源");
		if(writer != null){
			try {
				writer.close();
			} catch (IOException e) {
				e.printStackTrace();
			}
		}
		
		logger.info("=========>清理数据库连接资源");
		if(sqlUtil != null){
			sqlUtil.closeConn();
		}
		
		if(redisUtil != null){
			redisUtil.destroy();
			logger.info("Redis连接池关闭成功！");
		}
	}

}
